package groupOne.app.DWD.db;

import groupOne.app.BaseAppSql;
import groupOne.common.Constant;
import groupOne.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdCourseReviewInFo_lyy extends BaseAppSql {
    public static void main(String[] args) {

        new DwdCourseReviewInFo_lyy().init(
                3001,
                3,
                "Edu_Dwd_CourseReviewInfo"
        );

    }   @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //使用kafka连接器获取ods_db的动态表
        readOdsDb(tEnv,"Edu_Dwd_CourseReviewInfo");


        //从动态表读取课程评价表
        Table result = tEnv.sqlQuery("select " +
                " data['id'] id ," +
                " data['user_id'] user_id ," +
                " data['course_id'] course_id ," +
                " data['review_txt'] review_txt ," +
                " data['review_stars'] review_stars ," +
                " data['create_time'] create_time ," +
                " data['deleted'] deleted ," +
                " ts ts" +
                " from ods_db" +
                " where `database`='edu' " +
                " and `table`='review_info' " +
                " and `type`='insert'");

        tEnv.createTemporaryView("reviewInfo",result);

        // 定义动态表输出topic关联
        tEnv.executeSql("create table dwd_course_review_info(" +
                " id string," +
                " user_id string," +
                " course_id string," +
                " review_txt string," +
                " review_stars string," +
                " create_time string," +
                " deleted string," +
                " ts BIGINT" +
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_COURSE_REVIEW_INFO)
        );

        result.executeInsert("dwd_course_review_info");
            }
}
